package com.example.kafkademo.offsetcommit;

import org.apache.kafka.clients.consumer.CommitFailedException;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * @author yanyun zhu
 * @version 1.0
 * @date 2021/12/29 9:08
 *        生产中推荐使用方法
 */
public class ManualCommit {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test");
        props.put("enable.auto.commit", "false");    //默认true, 自动提交
//        props.put("auto.commit.interval.ms", "2000");     //表明 Kafka 每 5 秒会为你自动提交一次位移(默认)
        props.put("max.poll.records", 500);     // 默认每次拉取 500 条
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("test-a"));

        /**
         * commitAsync（）提交的时候同样会有失败的情况发生，那么我们应该怎么处理呢？读者有可能想到的是重试，问题的关键也就在这里了。如果某一次异步提交的消费位移为x ， 但是提交失败了，
         * 然后下一次又异步提交了消费位移为x+y，这次成功了。如果这里引入了重试机制，前一次的异步提交的消费位移在重试的时候提交成功了，那么此时的消费位移又变为了x 。如果此时发生异常（或者再均衡） ，
         * 那么恢复之后的消费者（或者新的消费者）就会从x 处开始消费消息，这样就发生了重复消费的问题。
         * 为此我们可以设置一个递增的序号来维护异步提交的顺序，每次位移提交之后就增加序号相对应的值。在遇到位移提交失败需要重试的时候，可以检查所提交的位移和序号的值的大小，如果前者小于后者，则说明有更大的位移己经提交了，
         * 不需要再进行本次重试：如果两者相同，则说明可以进行重试提交。除非程序编码错误，否则不会出现前者大于后者的情况。如果位移提交失败的情况经常发生，
         * 那么说明系统肯定出现了故障，在－般情况下，位移提交失败的情况很少发生，不重试也没有关系，后面的提交也会有成功的。重试会增加代码逻辑的复杂度，不重试会增加重复消费的概率。
         * 如果消费者异常退出，那么这个重复消费的问题就很难避免，因为这种情况下无法及时提交消费位移；如果消费者正常退出或发生再均衡的情况，那么可以在退出或再均衡执行之前使用同步提交的方式做最后的把关。
         */
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
//                process(records); // 处理消息
                consumer.commitAsync(); // 使用异步提交规避阻塞
//                consumer.commitAsync((offsets, exception) -> {
//                    if (exception != null){
//                        //                    handle(exception);
//                    }
//                });
            }
        } catch (Exception e) {
//            handle(e); // 处理异常
        } finally {
            try {
                /**
                 * 对于手动同步和异步提交结合的场景，如果poll出来的消息是500条，而业务处理200条的时候，业务抛异常了，后续消息根本就没有被遍历过，
                 * 如果调用没有参数的commit，那么提交的是500
                 * 看来还是需要有参数
                 */
                consumer.commitSync(); // 最后一次提交使用同步阻塞式提交
            } finally {
                consumer.close();
            }
        }
    }

}
